package per.zqx.rpc;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import lombok.SneakyThrows;
import per.zqx.utils.RabbitUtils;

import java.nio.charset.StandardCharsets;

/**
 * RpcServer,服务器端
 *
 * @author Sa  2021-06-02 21:37
 */
public class RpcServer {
    // 远程调用的消息队列
    static final String QUEUE_NAME = "rpc_queue_test";
    @SneakyThrows
    public static void main(String[] args) {
        try (Connection connection = RabbitUtils.getConnection()) {
            final Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            // 清空队列
            channel.queuePurge(QUEUE_NAME);
            // 限制消费
            channel.basicQos(1);
            System.out.println("Server. Ready for RPC Request ……");
            // TODO 加锁干嘛
            final Object monitor = new Object();
            // 消费者的回调函数
            DeliverCallback deliverCallback = (consumerTag, message) -> {
                // 通过 correlationId 作为唯一标识，构建资源，拿到返还数据的队列等信息
                final AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                        .Builder()
                        .correlationId(message.getProperties().getCorrelationId())
                        .build();
                // 消费的回调函数，进行业务处理，计算斐波拉契数列的第n值
                String resp = "";
                try {
                    String msg = new String(message.getBody(),"utf-8");
                    int n = Integer.parseInt(msg);
                    resp += fib(n);
                }catch (Exception e){
                    System.out.println(e.toString());
                }finally {
                    // 计算完毕后，将计算结果返还给调用者，correlationId + 调用者队列
                    channel.basicPublish("",
                            message.getProperties().getReplyTo(),
                            replyProps,
                            resp.getBytes(StandardCharsets.UTF_8));
                    // 手动确认
                    channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    System.out.println("计算结果 -> " + resp);
                    // RabbitMq consumer worker thread notifies the RPC server owner thread
                    // 消费者（这里指的调用者）通知唤醒服务端
                    synchronized (monitor){
                        monitor.notify();
                    }
                }
            };
            // 等待队列消息，消费消息，处理任务，不要自动确认
            channel.basicConsume(QUEUE_NAME,false,deliverCallback,consumerTag -> {});
            while (true){
                synchronized (monitor){
                    try {
                        monitor.wait();
                    }catch (InterruptedException e){
                        e.printStackTrace();
                    }
                }
            }

        }
    }

    private static int fib(int n) {
        if (n == 0) return 0;
        if (n == 1) return 1;
        return fib(n - 1) + fib(n - 2);
    }
}
